package com.cskaoyan.order.mq.consumer;

import com.cskaoyan.order.dal.persistence.OrderMapper;
import com.cskaoyan.order.service.OrderCoreService;
import com.cskaoyan.order.service.OrderQueryService;
import com.cskaoyan.order.service.impl.OrderCoreServiceImpl;
import com.mysql.cj.x.protobuf.MysqlxDatatypes;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.UnsupportedEncodingException;
import java.util.List;

/**
 * @Description:
 * @since: 2022-10-28 19:57
 * @author: lazy
 */
@Component
public class DelayOrderCancelConsumer {

    private DefaultMQPushConsumer consumer;

    @Autowired
    OrderQueryService orderQueryService;

    @Autowired
    OrderCoreService orderCoreService;


    //1、创建接受消费者消息的consumer对象
    @PostConstruct
    public void init() {
        consumer = new DefaultMQPushConsumer("order_delay_cancel_producer");
        //2、设置namesrv地址
        consumer.setNamesrvAddr("127.0.0.1:9876");
        //3、设置消息监听器
        consumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //1、实现订单超时自动取消的业务逻辑。
                //获取消息
                MessageExt message = msgs.get(0);
               // long startTime = Long.parseLong(message.getUserProperty("startTime"));
               // long end = System.currentTimeMillis();
                //获取消息体数据
                byte[] body = message.getBody();

                //业务逻辑处理数据
                try {
                    String orderId = new String(body, 0, body.length, "utf-8");
                    //2、根据orderId查询数据库中该ID的订单状态
                    Integer status = orderQueryService.queryStatusByOrderId(orderId);
                    //3、订单已支付则什么都不做
                    //4、订单未支付，执行自动取消的逻辑
                    if (status == 0) {
                        //5、修改订单的状态为已取消，访问tb_order_item查询订单中包含的所包含的订单商品条目，根据订单条目中商品购买数量更改数据库中相关数据
                        orderCoreService.cancelOrder(orderId);
                    }
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }
                return null;
            }
        });

        try {
            //4、订阅主题
            consumer.subscribe("delay_order", "*");
            //5、启动消费者
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}
